package rabbitmq.confirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认模式
 * 发布确认有3种策略，为单独确认、批量确认、异步确认
 */
public class PublisherConfirms {
    private static final Integer MESSAGE_COUNT = 200;
    public static Connection createConnection() throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前开放端口号
        connectionFactory.setUsername(Constants.USER_NAME);//账号
        connectionFactory.setPassword(Constants.PASSWORD);  //密码
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机
        return connectionFactory.newConnection();
    }

    public static void main(String[] args) throws Exception {
        // Strategy #1: Publishing Messages Individually
        // 单独确认
        publishingMessagesIndividually();

        //Strategy #2: Publishing Messages in Batches
        //批量确认
        publishingMessagesInBatches();

        //Strategy #3: Handling Publisher Confirms Asynchronously
        //异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    /**
     * 异步确认
     */
    private static void handlingPublisherConfirmsAsynchronously() throws Exception {
        try (Connection connection = createConnection()) {
            // 在 try 中已经建立了连接
            // 2、开启信道，也就是根据 连接创建信道
            Channel channel = connection.createChannel();
            // 使用默认的交换机，
            // 3、设置信道为confirm模式
            channel.confirmSelect();
            // 4、声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3,true,false,false,null);
            // 5、监听confirm
            //集合中存储的是未确认的消息ID，需要存放的顺序为有序的
            long start = System.currentTimeMillis();
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());

            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple) {
                        // confirmSet.headSet(n)⽅法返回当前集合中⼩于n的集合
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    }else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple) {
                        // confirmSet.headSet(n)⽅法返回当前集合中⼩于n的集合
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    }else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                    // 业务需要根据实际场景进行处理, 比如重发, 此处代码省略
                }
            });

            // 6、发送信息
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirms"+i;
                // 获取发送的信道的信息的ID
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());
                confirmSeqNo.add(seqNo);
            }
            while(!confirmSeqNo.isEmpty()) {
                // 当不为空的时候继续等待确认
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.printf("异步确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end - start);
        }
    }

    /**
     * 批量确认
     */
    private static void publishingMessagesInBatches() throws Exception {
        try (Connection connection = createConnection()) {
            // 在 try 中已经建立了连接
            // 2、开启信道，也就是根据 连接创建信道
            Channel channel = connection.createChannel();
            // 使用默认的交换机，
            // 3、设置信道为confirm模式
            channel.confirmSelect();
            // 4、声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2,true,false,false,null);
            // 5、发送消息,并等待确认
            long start = System.currentTimeMillis();
            int batchSize = 100;
            int outstandingMessageCount = 0;
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirms"+i;
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());
                // 每次发送信息之后，都进行将个数+1
                outstandingMessageCount++;
                if (outstandingMessageCount == batchSize) {
                    // 到达处理次数
                    // 等待确定，每次信息等待都是5s
                    channel.waitForConfirmsOrDie(5000);
                    outstandingMessageCount = 0;
                }
            }
            long end = System.currentTimeMillis();
            System.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end - start);
        }
    }

    /**
     * 单独确认
     */
    private static void publishingMessagesIndividually() throws Exception {
        try (Connection connection = createConnection()) {
            // 在 try 中已经建立了连接
            // 2、开启信道，也就是根据 连接创建信道
            Channel channel = connection.createChannel();
            // 使用默认的交换机，
            // 3、设置信道为confirm模式
            channel.confirmSelect();
            // 4、声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
            // 5、发送消息,并等待确认
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirms"+i;
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE1, null, msg.getBytes());
                // 等待确定，每次信息等待都是5s
                channel.waitForConfirmsOrDie(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end - start);
        }
    }
}
